package com.ydl.learning.flink.demo;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;

/**
 * @author ydl
 * @since 2021/10/21
 */
public class EsJavaSink extends RichSinkFunction<String> {
  private static RestHighLevelClient client = null;

  @Override
  public void open(Configuration parameters) throws Exception {
    final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
    credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("", ""));
    RestClientBuilder builder = RestClient.builder(new HttpHost("10.50.168.216", 9200))
        .setHttpClientConfigCallback((httpClientBuilder)
            -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
    client = new RestHighLevelClient(builder);
  }

  @Override
  public void close() throws Exception {
    client.close();
  }

  @Override
  public void invoke(String value, Context context) throws Exception {
    System.out.println(value);

  }
}
